/*
Copyright 2010 Bryce Alcock

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.com

 */
package com.ecocrypt.proxy.forwarder.remote;

import com.ecocrypt.proxy.forwarder.ForwardProxy;
import com.ecocrypt.proxy.forwarder.RemoteForwarderStub;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * This is the remote Service Request Class.  This class handles the remote presentation
 * of the GUI.  It will Manage the full duplex socket communication between the connected
 * FowardProxy and the client.  Note that each forward proxy can have multiple forwarders.
 *
 * @author bryce.alcock
 */
public class RemoteServiceRequest extends Thread {

    //  This will be incremented each time a new remote connection is established.
    private static final AtomicInteger serviceRequestId = new AtomicInteger();

    //  The input, output streams, alive status, and message queue.
    InputStream inputStream;
    OutputStream outputStream;
    Socket sock;
    boolean alive = true;
    LinkedBlockingQueue<Serializable> messageSendQueue = new LinkedBlockingQueue<Serializable>();

    // A reference to the incoming and outgoing Thread
    // so this thread can be inturrupted if need.
    RemoteServiceIncomingSocketStream incomingThread;
    Thread outgoingThread;

    public RemoteServiceRequest(Socket s) throws IOException {
        
        //  Get the streams from the socket.
        outputStream = s.getOutputStream();
        inputStream = s.getInputStream();

	sock=s;
        // Set the threads.
        this.setName("RemoteServiceRequest SenderThread: " + serviceRequestId.incrementAndGet());
        incomingThread = new RemoteServiceIncomingSocketStream(inputStream);

        //----------------------------------------------------------------------
        //  Add all the Existing Forwarders as stubs to this request.
        for (RemoteForwarderStub stub : ForwardProxy.remoteForwarderStubs) {
            messageSendQueue.add(stub);
        }
    }

    public void startThreads(){
        this.start();
        incomingThread.start();
    }
    private void terminateConnection(){

	alive=false;
	incomingThread.interrupt();
	this.interrupt();

	if(outputStream!=null){
	    try{
		outputStream.close();
	    }catch(Exception e){
		//  Bury it.
	    }

	}
	if(sock!=null){
	    if(!sock.isClosed()){
		try {
		    sock.close();
		} catch (IOException ex) {
		    Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
		}
	    }

	}

    }

    /**
     * Add a new message to send to the command console.
     * @param s
     */
    public void add(Serializable s){
	messageSendQueue.add(s);
    }

    @Override
    public void run() {

        AtomicInteger sendCount = new AtomicInteger();
        try {
            ObjectOutputStream oos = new ObjectOutputStream(outputStream);
            while (alive) {
                try {
                    Serializable s = messageSendQueue.take();
                    oos.writeObject(s);
		    oos.flush();
                    if(sendCount.incrementAndGet()%1==0){
                        oos.flush();
                        oos.reset();
                    }
                } catch (InterruptedException ex) {
                    validateAliveStatus();
                    continue;
                }
            }
        } catch (IOException ex) {
            Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
            return;
        }
    }

    // Currently if we get inturrupted, we will shut down.
    // We can add re-try logic on the client/command console if needed.
    private void validateAliveStatus() {
        //  This can set alive to false.
        alive = false;
    }

    /**
     * Inner class to handle incoming data from the client/command console.
     */
    private class RemoteServiceIncomingSocketStream extends Thread{

        InputStream inputStream ;
        RemoteServiceIncomingSocketStream(InputStream ois){
            inputStream = ois;
        }
        @Override
        public void run() {
	    ObjectInputStream ois = null;
	    try {
		ois = new ObjectInputStream(inputStream);
		while (alive) {
		    try {
			Object o = ois.readObject();
			processInput(o);
		    } catch (java.io.EOFException ex){
			alive = false;
			Logger.getLogger(this.getClass().getName()).log(Level.INFO,"Remote Host Shut Down");
			terminateConnection();
		    } catch (IOException ex) {
			Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
		    } catch (ClassNotFoundException ex) {
			Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
		    }
		}
	    } catch (IOException ex) {
		Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
	    } finally {
		try {
		    ois.close();
		} catch (IOException ex) {
		    Logger.getLogger(RemoteServiceRequest.class.getName()).log(Level.SEVERE, null, ex);
		}
	    }
        }
        private void processInput(Object o){
            // Currently, we are doing nothing with the
            // inbound requests.
            // TODO: Add handlers.
            System.out.println(String.valueOf(o));
        }
    }
}
